package org.apache.hadoop.hdfs.server.namenode.ha;

import static org.apache.hadoop.util.Time.monotonicNow;

import java.io.IOException;
import java.net.URI;
import java.net.URL;
import java.security.PrivilegedAction;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ha.ServiceFailedException;
import org.apache.hadoop.hdfs.DFSUtil;
import org.apache.hadoop.hdfs.HAUtil;
import org.apache.hadoop.hdfs.server.namenode.CheckpointConf;
import org.apache.hadoop.hdfs.server.namenode.FSImage;
import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
import org.apache.hadoop.hdfs.server.namenode.NNStorage.NameNodeFile;
import org.apache.hadoop.hdfs.server.namenode.NameNode;
import org.apache.hadoop.hdfs.server.namenode.SaveNamespaceCancelledException;
import org.apache.hadoop.hdfs.server.namenode.TransferFsImage;
import org.apache.hadoop.hdfs.util.Canceler;
import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.security.UserGroupInformation;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

/**
 *  这是一个后台线程，在standby  namenode里运行的，
 *  周期性的运行对namespace(元数据)的一次checkpoint操作
 *  当执行checkpoint操作的时候，它会把自己的namespace（内存里的元数据）写一份到磁盘文件上，fsimage文件，
 *  然后将这个fsimage文件，上传一份到active namenode，
 *
 *  active namenode就可以时不时的获取到一份最新的fsimage, 然后清空之前的一些edits log文件，
 *  它这样可疑保证每次namenode启动的时候，只要将最新的一份fsimage和少量的edits log进行合并即可
 */
@InterfaceAudience.Private
public class StandbyCheckpointer {
  private static final Log LOG = LogFactory.getLog(StandbyCheckpointer.class);
  private static final long PREVENT_AFTER_CANCEL_MS = 2*60*1000L;
  private final CheckpointConf checkpointConf;
  private final Configuration conf;
  private final FSNamesystem namesystem;
  private long lastCheckpointTime;
  // 核心组件
  private final CheckpointerThread thread;
  private final ThreadFactory uploadThreadFactory;
  private URL activeNNAddress;
  private URL myNNAddress;

  private final Object cancelLock = new Object();
  private Canceler canceler;
  
  // Keep track of how many checkpoints were canceled.
  // This is for use in tests.
  private static int canceledCount = 0;
  
  public StandbyCheckpointer(Configuration conf, FSNamesystem ns)
      throws IOException {
    this.namesystem = ns;
    this.conf = conf;
    this.checkpointConf = new CheckpointConf(conf); 
    this.thread = new CheckpointerThread();
    this.uploadThreadFactory = new ThreadFactoryBuilder().setDaemon(true)
        .setNameFormat("TransferFsImageUpload-%d").build();

    setNameNodeAddresses(conf);
  }

  /**
   * Determine the address of the NN we are checkpointing
   * as well as our own HTTP address from the configuration.
   * @throws IOException 
   */
  private void setNameNodeAddresses(Configuration conf) throws IOException {
    // Look up our own address.
    myNNAddress = getHttpAddress(conf);

    // Look up the active node's address
    Configuration confForActive = HAUtil.getConfForOtherNode(conf);
    activeNNAddress = getHttpAddress(confForActive);
    
    // Sanity-check.
    Preconditions.checkArgument(checkAddress(activeNNAddress),
        "Bad address for active NN: %s", activeNNAddress);
    Preconditions.checkArgument(checkAddress(myNNAddress),
        "Bad address for standby NN: %s", myNNAddress);
  }
  
  private URL getHttpAddress(Configuration conf) throws IOException {
    final String scheme = DFSUtil.getHttpClientScheme(conf);
    String defaultHost = NameNode.getServiceAddress(conf, true).getHostName();
    URI addr = DFSUtil.getInfoServerWithDefaultHost(defaultHost, conf, scheme);
    return addr.toURL();
  }
  
  /**
   * Ensure that the given address is valid and has a port
   * specified.
   */
  private static boolean checkAddress(URL addr) {
    return addr.getPort() != 0;
  }

  public void start() {
    LOG.info("Starting standby checkpoint thread...\n" +
        "Checkpointing active NN at " + activeNNAddress + "\n" +
        "Serving checkpoints at " + myNNAddress);
    thread.start();
  }
  
  public void stop() throws IOException {
    cancelAndPreventCheckpoints("Stopping checkpointer");
    thread.setShouldRun(false);
    thread.interrupt();
    try {
      thread.join();
    } catch (InterruptedException e) {
      LOG.warn("Edit log tailer thread exited with an exception");
      throw new IOException(e);
    }
  }

  public void triggerRollbackCheckpoint() {
    thread.interrupt();
  }

  /**
   * 整个检查点执行操作的逻辑都是在doCheckpoint()方法中实现的。
   *
   * 1. 首先获取当前保存的fsimage的prevCheckpointTxId，
   * 2. 然后获取最近更新的editlog的thisCheckpointTxId，只有新的thisCheckpointTxId大于prevCheckpointTxId，
   *     也就是当前命名空间有更新， 但是并没有保存到新的fsimage文件中时，才有必要进行一次检查点操作。
   * 3. 判断完成后，doCheckpoint()会调用saveNamespace()方法将最新的命名空间保存到fsimage文件中。
   * 4. 之后构造一个线程，将新产生的fsimage文件通过HTTP方式上传到AvtiveNamenode中。
   */
  private void doCheckpoint() throws InterruptedException, IOException {
    assert canceler != null;
    final long txid;
    final NameNodeFile imageType;
    
    namesystem.longReadLockInterruptibly();
    try {
      assert namesystem.getEditLog().isOpenForRead() :
        "Standby Checkpointer should only attempt a checkpoint when NN is in standby mode, but the edit logs are in an unexpected state";

      // 获取当前  Standby Namenode 上保存的最新的fsimage对象
      FSImage img = namesystem.getFSImage();
      // 获取 fsimage 中保存的txid, 也就是上一次进行检查点操作时保存的txid
      long prevCheckpointTxId = img.getStorage().getMostRecentCheckpointTxId();
      // 获取当前命名空间的最新txid, 也就是收到的editlog的最新的txid
      long thisCheckpointTxId = img.getLastAppliedOrWrittenTxId();

      // thisCheckpointTxId 一定大于 prevCheckpointTxId
      assert thisCheckpointTxId >= prevCheckpointTxId;
      // 如果相等则没有必要执行检查点操作，当前fsimage已经时最新的了
      if (thisCheckpointTxId == prevCheckpointTxId) {
        LOG.info("A checkpoint was triggered but the Standby Node has not received any transactions since the last checkpoint at txid " + thisCheckpointTxId + ". Skipping...");
        return;
      }

      if (namesystem.isRollingUpgrade() && !namesystem.getFSImage().hasRollbackFSImage()) {
        // 如果当前namenode正在执行升级操作，则创建 fsiamge_rollback文件
        imageType = NameNodeFile.IMAGE_ROLLBACK;
      } else {
        // 在正常情况下创建 fsimage文件
        imageType = NameNodeFile.IMAGE;
      }
      // 调用FSImage.saveNamespace()方法，将内存中的元数据保存到磁盘文件上去
      // fsimage_00000000000001729,跟上 txid
      // 标识了说，这个fsimage是已经合并了 txid=1729 的所有edits log
      img.saveNamespace(namesystem, imageType, canceler); // TODO 核心代码
      txid = img.getStorage().getMostRecentCheckpointTxId();
      assert txid == thisCheckpointTxId : "expected to save checkpoint at txid=" + thisCheckpointTxId + " but instead saved at txid=" + txid;

      // Save the legacy OIV image, if the output dir is defined.
      String outputDir = checkpointConf.getLegacyOivImageDir();
      if (outputDir != null && !outputDir.isEmpty()) {
        img.saveLegacyOIVImage(namesystem, outputDir, canceler);
      }
    } finally {
      namesystem.longReadUnlock();
    }

    //  用另外一个线程 异步的将这个刚刚写好的fsimage文件 通过 HTTP 传输到active namenode上去
    ExecutorService executor = Executors.newSingleThreadExecutor(uploadThreadFactory);
    Future<Void> upload = executor.submit(new Callable<Void>() {
      @Override
      public Void call() throws IOException {
        // 法将新生成的fsimage文件上传到Active Namenode中了
        TransferFsImage.uploadImageFromStorage(activeNNAddress, conf, namesystem.getFSImage().getStorage(), imageType, txid, canceler); // TODO 核心代码
        return null;
      }
    });
    executor.shutdown();
    try {
      upload.get();
    } catch (InterruptedException e) {
      // The background thread may be blocked waiting in the throttler, so interrupt it.
      upload.cancel(true);
      throw e;
    } catch (ExecutionException e) {
      throw new IOException("Exception during image upload: " + e.getMessage(), e.getCause());
    }
  }
  
  /**
   * Cancel any checkpoint that's currently being made,
   * and prevent any new checkpoints from starting for the next
   * minute or so.
   */
  public void cancelAndPreventCheckpoints(String msg) throws ServiceFailedException {
    thread.preventCheckpointsFor(PREVENT_AFTER_CANCEL_MS);
    synchronized (cancelLock) {
      // Before beginning a checkpoint, the checkpointer thread
      // takes this lock, and creates a canceler object.
      // If the canceler is non-null, then a checkpoint is in
      // progress and we need to cancel it. If it's null, then
      // the operation has not started, meaning that the above
      // time-based prevention will take effect.
      if (canceler != null) {
        canceler.cancel(msg);
      }
    }
  }
  
  @VisibleForTesting
  static int getCanceledCount() {
    return canceledCount;
  }

  private long countUncheckpointedTxns() {
    FSImage img = namesystem.getFSImage();
    return img.getLastAppliedOrWrittenTxId() -
      img.getStorage().getMostRecentCheckpointTxId();
  }


  private class CheckpointerThread extends Thread {
    private volatile boolean shouldRun = true;
    private volatile long preventCheckpointsUntil = 0;

    private CheckpointerThread() {
      super("Standby State Checkpointer");
    }
    
    private void setShouldRun(boolean shouldRun) {
      this.shouldRun = shouldRun;
    }

    @Override
    public void run() {
      SecurityUtil.doAsLoginUserOrFatal(
          new PrivilegedAction<Object>() {
          @Override
          public Object run() {
            doWork(); // TODO 核心代码
            return null;
          }
        });
    }

    /**
     * Prevent checkpoints from occurring for some time period
     * in the future. This is used when preparing to enter active
     * mode. We need to not only cancel any concurrent checkpoint,
     * but also prevent any checkpoints from racing to start just
     * after the cancel call.
     * 
     * @param delayMs the number of MS for which checkpoints will be
     * prevented
     */
    private void preventCheckpointsFor(long delayMs) {
      preventCheckpointsUntil = monotonicNow() + delayMs;
    }

    /**
     * StandbyNamenode会持有一个StandbyCheckpointer类，这个类维护着一个叫作CheckpointerThread的线程，
     * CheckpointerThread线程会每隔1000*Math.min(checkpointCheck Period, checkpointPeriod)秒检测是否执行一次检查点逻辑
     * （checkpointCheckPeriod由dfs.namenode.checkpoint.period配置，checkpointPeriod则由dfs.namenode.checkpoint.check.period配置）。
     * 整个检查点逻辑是在CheckpointerThread.doWork()方法中实现的，
     *
     * doWork()方法首先会判断是否满足检查点操作的两个条件，如果满足则调用doCheckpoint()执行检查点操作。
     */
    private void doWork() {
      final long checkPeriod = 1000 * checkpointConf.getCheckPeriod(); // 60s
      // 重置checkpoint时间，这样我们就不会总是在启动时设置checkpoint。
      lastCheckpointTime = monotonicNow();
      while (shouldRun) {
        boolean needRollbackCheckpoint = namesystem.isNeedRollbackFsImage();
        if (!needRollbackCheckpoint) {
          try {
            // 间隔时长 默认 60s
            Thread.sleep(checkPeriod);
          } catch (InterruptedException ie) {
          }
          if (!shouldRun) {
            break;
          }
        }
        try {
          // We may have lost our ticket since last checkpoint, log in again, just in case
          if (UserGroupInformation.isSecurityEnabled()) {
            UserGroupInformation.getCurrentUser().checkTGTAndReloginFromKeytab();
          }
          
          final long now = monotonicNow();
          // 获得最后一次往journalNode 写入的 txid 和最近一次做检查点的txid的差值
          final long uncheckpointed = countUncheckpointedTxns();
          // 计算当前时间和上一次检查点操作时间的间隔
          final long secsSinceLast = (now - lastCheckpointTime) / 1000;
          
          boolean needCheckpoint = needRollbackCheckpoint;

          if (needCheckpoint) {
            LOG.info("Triggering a rollback fsimage for rolling upgrade.");
          }
          // 如果 满足第1个条件，就可以执行checkpoint操作
          // uncheckpointd(还没有合并到fsimage的edits log的数量) >= 100万
          // fsimage文件如果落后edits log已经有100万条数据了，那么此时就必须得执行一次checkpoint操作
          else if (uncheckpointed >= checkpointConf.getTxnCount()) { // 100万
            LOG.info("Triggering checkpoint because there have been " +  uncheckpointed + " txns since the last checkpoint, which " + "exceeds the configured threshold " + checkpointConf.getTxnCount());
            needCheckpoint = true;
          }
          // secsSinceLast (当前时间到上一次checkpoint的间隔) >= 1小时（3600秒）
          // 如果距离上一次checkpoint操作超过了1个小时，那么此时就必须执行一次checkpoint操作
          else if (secsSinceLast >= checkpointConf.getPeriod()) {
            LOG.info("Triggering checkpoint because it has been " + secsSinceLast + " seconds since the last checkpoint, which " + "exceeds the configured interval " + checkpointConf.getPeriod());
            needCheckpoint = true;
          }

          // 执行checkpoint 操作，就2个条件：
          // 第1个条件：如果超过100万个edits log没合并到fsimage里去，就执行
          // 第2个条件：如果超过1小时，那么就执行一次
          synchronized (cancelLock) {
            if (now < preventCheckpointsUntil) {
              LOG.info("But skipping this checkpoint since we are about to failover!");
              canceledCount++;
              continue;
            }
            assert canceler == null;
            canceler = new Canceler();
          }

          // 如果需要执行checkpoint
          if (needCheckpoint) {
            doCheckpoint(); // TODO 核心代码

            // reset needRollbackCheckpoint to false only when we finish a ckpt for rollback image
            if (needRollbackCheckpoint && namesystem.getFSImage().hasRollbackFSImage()) {
              namesystem.setCreatedRollbackImages(true);
              namesystem.setNeedRollbackFsImage(false);
            }
            lastCheckpointTime = now;
          }
        } catch (SaveNamespaceCancelledException ce) {
          LOG.info("Checkpoint was cancelled: " + ce.getMessage());
          canceledCount++;
        } catch (InterruptedException ie) {
          LOG.info("Interrupted during checkpointing", ie);
          continue;
        } catch (Throwable t) {
          LOG.error("Exception in doCheckpoint", t);
        } finally {
          synchronized (cancelLock) {
            canceler = null;
          }
        }
      }
    }
  }

  @VisibleForTesting
  URL getActiveNNAddress() {
    return activeNNAddress;
  }
}
